package com.zzm.controller;

import java.util.List;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.ResponseBody;
import org.springframework.web.bind.annotation.RestController;

import com.alibaba.rocketmq.client.Validators;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.client.producer.DefaultMQProducer;
import com.alibaba.rocketmq.client.producer.LocalTransactionExecuter;
import com.alibaba.rocketmq.client.producer.LocalTransactionState;
import com.alibaba.rocketmq.client.producer.SendResult;
import com.alibaba.rocketmq.client.producer.SendStatus;
import com.alibaba.rocketmq.client.producer.TransactionExecuter;
import com.alibaba.rocketmq.client.producer.TransactionMQProducer;
import com.alibaba.rocketmq.client.producer.TransactionSendResult;
import com.alibaba.rocketmq.common.message.Message;
import com.alibaba.rocketmq.common.message.MessageAccessor;
import com.alibaba.rocketmq.common.message.MessageConst;
import com.alibaba.rocketmq.common.message.MessageQueue;
import com.alibaba.rocketmq.example.transaction.TransactionExecuterImpl;

/**
 * @Author:钟志苗
 * @Date: 8:36 2017/5/13
 * @Description:
 */
@RestController
public class ProducerController {
	private static Logger log = LoggerFactory.getLogger(ProducerController.class);

	/**
	 * 普通消息生产者
	 */
	@Autowired
	@Qualifier(value = "default")
	private DefaultMQProducer defaultProducer;

	/**
	 * 事务消息生产者
	 */
	@Autowired
	@Qualifier(value = "trans")
	private TransactionMQProducer transactionProducer;
	

	/**
	 * 发送顺序消息
	 * 
	 * 
	 * https://www.jianshu.com/p/453c6e7ff81c
	 * 
	 * @throws Exception
	 */
	@RequestMapping(value = "/sendMsg", method = RequestMethod.GET)
	public void sendMsg() throws Exception {
		for (int orderId = 0; orderId < 3; orderId++) {
			for (int j = 0; j < 3; j++) {
				String body = orderId + ":" + j;
				Message msg = new Message("order", "TagA", "OrderID11113", body.getBytes());

				// Object arg=orderId与消息队列选择器一起工作的参数，同一订单号会分配到同一队列
				defaultProducer.send(msg, (List<MessageQueue> mqs, Message msg1, Object arg) -> {

					int value = arg.hashCode();
					System.out.print(value + " ");
					if (value < 0) {
						value = Math.abs(value);
					}
					System.out.print(value + "%" + mqs.size() + "=");
					value = value % mqs.size();
					System.out.print(value + "");
					System.out.println("MessageQueueIndex-" + value + ",body=" + body + ",msg=" + msg);
					return mqs.get(value);
				}, orderId);
			}

		}

	}

	/**
	 * 普通消息
	 * 
	 * Producer轮询某topic下的所有队列的方式来实现发送方的负载均衡
	 * 
	 * @param tag
	 * @param body
	 * @param total
	 * @return
	 * @throws Exception
	 */
	@RequestMapping("/rocketMq/{tag}/{body}/{total}")
	@ResponseBody
	public String rocketMq(@PathVariable String tag, @PathVariable String body, @PathVariable Integer total)
			throws Exception {
		long t1 = System.currentTimeMillis();
		SendResult sendResult = null;
		int fail = 0;
		for (int i = 1; i < total + 1; i++) {
			Message msg = new Message("test", // topic
					"TagA", // tag
					"OrderID11112", // key
					(i + "").getBytes("utf-8"));// body

			// 发送消息并返回结果 Producer轮询某topic下的所有队列的方式来实现发送方的负载均衡
			sendResult = defaultProducer.send(msg);

			if (sendResult.getSendStatus().name().equals(SendStatus.SEND_OK.name())) {
				fail++;
			} else {
				System.out.println(sendResult.getSendStatus().name());
			}
		}
		long t2 = System.currentTimeMillis();
		return "所花时间：" + (t2 - t1) + "成功次数" + fail;
	}
	
	TransactionExecuter tranExecuter = new TransactionExecuterImpl();

	/**
	 * 发送事务消息
	 * 
	 * @param id
	 * @param arg1
	 * @return
	 */
	@RequestMapping(value = "/sendTransactionMsg/{id}/{arg1}", method = RequestMethod.GET)
	public String sendTransactionMsg(@PathVariable String id, @PathVariable String arg1) {
		SendResult sendResult = null;
		try {
			// 构造消息
			Message msg = new Message("test", // topic
					"TagA", // tag
					"OrderID11111", // key
					(id + "").getBytes("utf-8"));// body

//			sendResult = transactionProducer.sendMessageInTransaction(msg, (Message msg1, Object arg) -> {
			/*sendResult = transactionProducer.sendMessageInTransaction(msg, (Message msg1, Object arg) -> {
				// 发送事务消息，LocalTransactionExecute的executeLocalTransactionBranch方法中执行本地逻辑
				// sendResult = sendMessageInTransaction(msg, (Message msg1, Object arg) -> {
				int value = Integer.valueOf(arg.toString());
				System.out.println("value=" + value + ", 执行本地事务(结合自身的业务逻辑)。。。完成");
				if (value == 0) {
					throw new RuntimeException("Could not find db");
				} else if ((value % 5) == 0) {
					return LocalTransactionState.ROLLBACK_MESSAGE;
				} else if ((value % 4) == 0) {
					return LocalTransactionState.COMMIT_MESSAGE;
				}
				return LocalTransactionState.ROLLBACK_MESSAGE;
			}, arg1);*/
			
			sendResult = sendMessageInTransaction(msg, (Message msg1, Object arg) -> {
				System.out.println("本地事务：" + LocalTransactionState.COMMIT_MESSAGE);
				return LocalTransactionState.COMMIT_MESSAGE;
			}, null);
			
			
			//sendResult = transactionProducer.sendMessageInTransaction(msg, tranExecuter, null);
			System.out.println(sendResult);
		} catch (Exception e) {
			e.printStackTrace();
		}
		return sendResult.toString();
	}

	/**
	 * 模拟无法发送事务确认（测试事务回查）
	 *
	 * 源码解读https://github.com/YunaiV/Blog/blob/master/RocketMQ/1011-RocketMQ%E6%BA%90%E7%A0%81%E8%A7%A3%E6%9E%90%EF%BC%9A%E4%BA%8B%E5%8A%A1%E6%B6%88%E6%81%AF.md#
	 * 
	 * @param msg
	 * @param tranExecuter
	 * @param arg
	 * @return
	 * @throws MQClientException
	 */
	public TransactionSendResult sendMessageInTransaction(final Message msg,
			final LocalTransactionExecuter tranExecuter, final Object arg) throws MQClientException {
		if (null == tranExecuter) {
			throw new MQClientException("tranExecutor is null", null);
		}
		Validators.checkMessage(msg, transactionProducer);

		// 发送事务消息
		SendResult sendResult = null;
		MessageAccessor.putProperty(msg, MessageConst.PROPERTY_TRANSACTION_PREPARED, "true");
		MessageAccessor.putProperty(msg, MessageConst.PROPERTY_PRODUCER_GROUP, transactionProducer.getProducerGroup());
		try {
			sendResult = transactionProducer.send(msg);
			System.out.println("事务消息msg=" + msg);
		} catch (Exception e) {
			throw new MQClientException("send message Exception", e);
		}

		// 处理发送【Half消息】结果
		LocalTransactionState localTransactionState = LocalTransactionState.UNKNOW;
		switch (sendResult.getSendStatus()) {
		// 如果消息发送成功，处理与消息关联的本地事务单元
		case SEND_OK: {
			try {
				if (sendResult.getTransactionId() != null) {
					msg.putUserProperty("__transactionId__", sendResult.getTransactionId());
				}
				// 执行【本地事务】逻辑
				localTransactionState = tranExecuter.executeLocalTransactionBranch(msg, arg);
				if (null == localTransactionState) {
					localTransactionState = LocalTransactionState.UNKNOW;
				}

				if (localTransactionState != LocalTransactionState.COMMIT_MESSAGE) {
					log.info("executeLocalTransactionBranch return {}", localTransactionState);
					log.info(msg.toString());
				}
			} catch (Throwable e) {
				log.info("executeLocalTransactionBranch exception", e);
				log.info(msg.toString());
			}
		}
			break;
			// 发送【Half消息】失败，标记【本地事务】状态为回滚
		case FLUSH_DISK_TIMEOUT:
		case FLUSH_SLAVE_TIMEOUT:
		case SLAVE_NOT_AVAILABLE:
			localTransactionState = LocalTransactionState.ROLLBACK_MESSAGE;
			break;
		default:
			break;
		}

		// 测试事务回查
		/*
		 * 结束事务：提交消息 COMMIT / ROLLBACK
		 * 
		 * 如果endTransaction方法执行失败，数据没有发送到broker，导致事务消息的
		 * 状态更新失败，broker会有回查线程定时（默认1分钟）扫描每个存储事务状态的表格文件，如果是已经提交或者回滚的消息直接跳过，
		 * 如果是prepared状态则会向Producer发起CheckTransaction请求，
		 * Producer会调用DefaultMQProducerImpl.checkTransactionState()
		 * 方法来处理broker的定时回调请求，
		 * 而checkTransactionState会调用我们的事务设置的决断方法来决定是回滚事务还是继续执行，
		 * 最后调用endTransactionOneway让broker来更新消息的最终状态。
		 */

		// this.endTransaction(sendResult, localTransactionState, localException);
		System.out.println("--模拟消息确认丢失--");

		// 返回【事务发送结果】
		TransactionSendResult transactionSendResult = new TransactionSendResult();
		transactionSendResult.setSendStatus(sendResult.getSendStatus());
		transactionSendResult.setMessageQueue(sendResult.getMessageQueue());
		transactionSendResult.setMsgId(sendResult.getMsgId());
		transactionSendResult.setQueueOffset(sendResult.getQueueOffset());
		transactionSendResult.setTransactionId(sendResult.getTransactionId());
		transactionSendResult.setLocalTransactionState(localTransactionState);
		return transactionSendResult;
	}
}
